package cc.xpcas.raft.role;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cc.xpcas.raft.LogEntry;
import cc.xpcas.raft.context.Context;
import cc.xpcas.raft.message.AppendEntriesRequest;
import cc.xpcas.raft.message.MessageContext;
import cc.xpcas.raft.message.RequestVoteRequest;
import cc.xpcas.raft.message.RequestVoteResponse;

/**
 * @author xp
 */
public class Candidate extends Role {

    private static final Logger LOG = LoggerFactory.getLogger(Candidate.class);

    private static final Random random = new Random();

    private final AtomicInteger votes = new AtomicInteger(0);

    private final AtomicBoolean timeoutFlag = new AtomicBoolean(false);

    private Future<?> future;

    public Candidate(Context context) {
        super(context);
    }

    /**
     * in mainline
     */
    @Override
    public void start() {
        LOG.info(String.format("%s: candidate start", getConfiguration().getServerId()));

        getContext().setVotedFor(getConfiguration().getServerId());
        requestVotes();
    }

    /**
     * in mainline
     */
    @Override
    public void end() throws InterruptedException {
        timeoutFlag.set(true);
        if (future != null) {
            future.cancel(false);
        }
    }

    private long genRandomVotingTimeout() {
       return getConfiguration().getVotingTimeout() + random.nextInt(2000);
    }

    /**
     * in mainline
     */
    private void requestVotes() {
        if (!getContext().isCandidate()) return;

        LOG.info(String.format("%s: request votes", getConfiguration().getServerId()));

        long term = getContext().incrementTerm();
        votes.set(1); // itself

        timeoutFlag.set(false);
        future = getContext().schedule(this::votingTimeoutHandler, genRandomVotingTimeout());

        RequestVoteRequest message = new RequestVoteRequest();
        message.setServerId(getConfiguration().getServerId());
        message.setTerm(term);
        LogEntry lastLog = getContext().getLastLog();
        if (lastLog != null) {
            message.setLastLogId(lastLog.getId());
            message.setLastLogTerm(lastLog.getTerm());
        } else {
            message.setLastLogId(0);
            message.setLastLogTerm(0);
        }

        getContext().getPeers().values().forEach(peer -> peer.send(message));
    }

    /**
     * in mainline
     */
    @Override
    public boolean handleRequestVoteResponse(MessageContext messageContext) {
        RequestVoteResponse message = messageContext.getMessage();
        if (message.isVoteGranted()) {
            int votes = this.votes.incrementAndGet();
            if (votes > getConfiguration().getNodeNum() / 2) {
                if (!timeoutFlag.compareAndSet(false, true)) {
                    // flag has already been set, means we have timed out
                    return true;
                }
                // elected
                LOG.info(String.format("%s: elected", getConfiguration().getServerId()));
                getContext().becomeLeader();
            }
        } else {
            getContext().catchUpTerm(message.getCurrentTerm());
        }
        return true;
    }

    /*
     * -1 new leader
     *  0 another one elected
     *  1 old leader
     */
    private boolean checkValidLeader(long term) {
        int v = getContext().catchUpTerm(term);
        if (v == 0) {
            getContext().becomeFollower();
        }
        return v <= 0;
    }

    /**
     * in mainline
     */
    @Override
    public boolean handleAppendEntriesRequest(MessageContext messageContext) throws InterruptedException {
        AppendEntriesRequest message = messageContext.getMessage();
        if (checkValidLeader(message.getTerm())) {
            return false;
        }
        return true;
    }

    /**
     * in scheduled
     */
    private void votingTimeoutHandler() {
        if (!timeoutFlag.compareAndSet(false, true)) {
            // flag has already been set, means we have been elected
            return;
        }
        // timeout
        LOG.info(String.format("%s: voting timeout", getConfiguration().getServerId()));
        getContext().runOnMainline(this::requestVotes);
    }
}
